package com.mk.es.service.transport.pool;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;

/**
 * EsTransportClient工厂类
 */
public class EsTransportClientFactory implements PooledObjectFactory<TransportClient> {
    Logger logger = LoggerFactory.getLogger(EsTransportClientFactory.class);

    /**
     * elasticsearch集群名称
     */
    private String esClusterName;

    private Boolean esClientTransportSniff;

    /**
     * elasticsearch服务器地址，多个,隔开
     */
    private String esClusterHosts;

    public EsTransportClientFactory(String esClusterName, Boolean esClientTransportSniff, String esClusterHosts) {
        this.esClusterName = esClusterName;
        this.esClientTransportSniff = esClientTransportSniff;
        this.esClusterHosts = esClusterHosts;
    }

    /**
     * 创建一个新TransportClient对象,并交付给对象池管理
     *
     * @return
     * @throws Exception
     */
    @Override
    public PooledObject<TransportClient> makeObject() throws Exception {
        Settings.Builder builder = Settings.builder().put("client.transport.ignore_cluster_name", true);

        //集群名称，elasticsearch默认的集群名称为elasticsearch
        if (StringUtils.isNoneBlank(esClusterName)) {
            builder.put("cluster.name", esClusterName);
        }
        if (esClientTransportSniff != null) {
            builder.put("client.transport.sniff", esClientTransportSniff);
        }

        //处理Elasticsearch连接地址，格式192.168.12.2:9300,192.168.12.1:9400
        List<InetSocketTransportAddress> addressList = new ArrayList<InetSocketTransportAddress>();
        if (StringUtils.isNoneBlank(esClusterHosts)) {
            String[] esClusterHostArr = StringUtils.split(esClusterHosts, ",");
            for (String hostStr : esClusterHostArr) {
                if (StringUtils.isBlank(hostStr)) {
                    continue;
                }
                String[] hostArr = StringUtils.split(hostStr, ":");
                if (hostArr.length != 2) {
                    continue;
                }

                String host = hostArr[0]; //ip地址
                int port = StringUtils.isNoneBlank(hostArr[1]) ? Integer.parseInt(hostArr[1]) : 9300; //默认地址9300
                addressList.add(new InetSocketTransportAddress(InetAddress.getByName(host), port));
            }
        }

        if (addressList.isEmpty()) {
            throw new IllegalArgumentException("ElasticSearch argument “es.cluster.hosts” is empty ");
        }

        //创建TransportClient连接对象
        TransportClient transportClient = new PreBuiltTransportClient(builder.build());
        transportClient.addTransportAddresses(addressList.toArray(new InetSocketTransportAddress[addressList.size()]));

        return new DefaultPooledObject<TransportClient>(transportClient);
    }

    /**
     * 销毁TransportClient对象
     *
     * @param pooledObject
     * @throws Exception
     */
    @Override
    public void destroyObject(PooledObject<TransportClient> pooledObject) throws Exception {
        TransportClient transportClient = pooledObject.getObject();
        if (transportClient != null) {
            //关闭连接
            transportClient.close();
        }
    }

    /**
     * 检测对象是否"有效"
     *
     * @param pooledObject
     * @return
     */
    @Override
    public boolean validateObject(PooledObject<TransportClient> pooledObject) {
        if (pooledObject == null) {
            return false;
        }

        TransportClient client = pooledObject.getObject();
        if (client == null) {
            return false;
        }
        return true;
    }

    /**
     * "激活"对象
     *
     * @param pooledObject
     * @throws Exception
     */
    @Override
    public void activateObject(PooledObject<TransportClient> pooledObject) throws Exception {
        //
    }

    /**
     * "钝化"对象
     *
     * @param pooledObject
     * @throws Exception
     */
    @Override
    public void passivateObject(PooledObject<TransportClient> pooledObject) throws Exception {
        //
    }
}
